package com.yanqu.road.connect;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import com.yanqu.road.message.YanQuMessage;
import com.yanqu.road.task.FlushAgentServerConnectionTask;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import io.netty.channel.Channel;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

public class AgentServerConnection {

	private static Logger logger = LogManager.getLogger(AgentServerConnection.class.getName());
	private static final int WRITE_LIMIT_MAX = 128;
	private static final int WRITE_LIMIT_MIN = 64;

	private boolean canWrite;
	protected Channel channel;
	private GenericFutureListener<Future<? super Void>> listener;
	private Runnable flushTask;
	private AtomicInteger counter;
	private BlockingQueue<YanQuMessage> messageQueue;
	private AtomicBoolean flushSwitch;


	public AgentServerConnection(Channel channel) {
		this.channel = channel;
		counter = new AtomicInteger();
		messageQueue = new ArrayBlockingQueue<>(1024);
		flushTask = new FlushAgentServerConnectionTask(this);
		flushSwitch = new AtomicBoolean(true);
		listener = future -> {
			counter.decrementAndGet();
			if (freshAndGetWriteTerminal()) {
				flush();
			}
		};
	}

	public void write(YanQuMessage msg) {
		boolean oResult = messageQueue.offer(msg);
		if(!oResult){
			logger.error("offer messageQueue(agent) fail,code:{},userId:{}", msg.getCode(), msg.getPlayerId());
		}
//        try {
//            messageQueue.put(msg);
//        } catch (InterruptedException e) {
//			logger.error("put messageQueue error,code:{},userId:{},:", msg.getCode(), msg.getPlayerId(), e);
//        }
        if (freshAndGetWriteTerminal()) {
			flush();
		}
	}

	public boolean isAlive() {
		try {
			if (channel == null) {
				return false;
			}
			return channel.isActive();
		} catch (Exception e) {
			logger.error("channel exception:", e);
			return false;
		}
	}

	public void flush() {
		if (messageQueue.isEmpty()) {
			return;
		}

		if (flushSwitch.getAndSet(false)) {
			EventExecutor executor = channel.eventLoop();
			executor.execute(flushTask);
		}
	}

	public void close() {
		try {
			channel.close();
		} catch (Exception e) {
			logger.error("", e);
		}
	}

	private boolean freshAndGetWriteTerminal() {
		if (canWrite) {
			if (counter.get() >= WRITE_LIMIT_MAX) {
				canWrite = false;
			}
		} else {
			if (counter.get() <= WRITE_LIMIT_MIN) {
				canWrite = true;
			}
		}
		return canWrite;
	}

	public Channel getChannel() {
		return channel;
	}

	public BlockingQueue<YanQuMessage> getMessageQueue() {
		return messageQueue;
	}

	public AtomicInteger getCounter() {
		return counter;
	}

	public AtomicBoolean getFlushSwitch() {
		return flushSwitch;
	}

	public GenericFutureListener<Future<? super Void>> getListener() {
		return listener;
	}
}
